package com.ytktt.study.week12.task6;

import lombok.Getter;

import java.util.concurrent.TimeUnit;

/**
 * TODO
 *
 * @author ytk
 */
public class KmqConsumer<T> {

    private final KmqBroker<T> broker;

    private Kmq<T> kmq;

    @Getter
    private String name;

    public KmqConsumer(KmqBroker<T> broker, String name) {
        this.broker = broker;
        this.name = name;
    }

    public void subscribe(String topic) {
        this.kmq = this.broker.findKmq(topic);
        if (null == this.kmq) {
            throw new RuntimeException("Topic[" + topic + "] doesn't exist.");
        }
        this.kmq.registerConsumer(name);
    }

    public KmqMessage<T> poll(long timeout, TimeUnit timeUnit) {
        return this.kmq.poll(this.name, timeout, timeUnit);
    }

    public void seekEarliest() {
        this.seek(0);
    }

    public void seek(int offset) {
        this.kmq.seek(name, offset);
    }

}
